package com.kuafu.latte.transport.client;

import com.kuafu.latte.exchange.DefaultPackage;
import com.kuafu.latte.exchange.Message;
import com.kuafu.latte.transport.MessageDecoder;
import com.kuafu.latte.transport.MessageEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.util.concurrent.DefaultThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.SocketAddress;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

public class NettyClient{

    private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);

    /**
     * daemon true 守护线程 false 用户线程
     */
    private static final NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup(4, new DefaultThreadFactory("NettyClientWorker", true));

    private Bootstrap bootstrap;

    private SocketAddress address;

    private volatile Channel channel; // volatile, please copy reference to use

    private int timeout;

    public NettyClient(final SocketAddress address,int timeout){
        this.address = address;
        if(timeout < 3000){
            this.timeout = 3000;
        }else{
            this.timeout = timeout;
        }
    }

    public void doOpen() throws Exception {
        bootstrap = new Bootstrap();
        bootstrap.group(nioEventLoopGroup)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, timeout)
                .channel(NioSocketChannel.class);

        bootstrap.handler(new ChannelInitializer() {

            protected void initChannel(Channel ch) throws Exception {
                //获取管道
                ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
                        .addLast("frameDecoder", new LengthFieldBasedFrameDecoder(20*1024, 0, 4, 0, 4))//
                        .addLast("frameEncoder", new LengthFieldPrepender(4))//
                        .addLast("decoder", new MessageDecoder())
                        .addLast("encoder", new MessageEncoder())
                        .addLast("handler", new NettyClientHandler(NettyClient.this));
            }
        });
    }

    public void doConnect() throws Exception {
        //.sync() 阻塞线程，直到连接上
        ChannelFuture future = bootstrap.connect(address);
        future.addListener(new ConnectionListener(this));
        /*boolean ret = future.awaitUninterruptibly(timeout, TimeUnit.MILLISECONDS);
        if (ret && future.isSuccess()) {
            Channel newChannel = future.channel();
            // 关闭旧的连接
            Channel oldChannel = channel; // copy reference
            if (oldChannel != null) {
                oldChannel.close();
            }
            channel = newChannel;
        } else if (future.cause() != null) {
            logger.error("connect failed");
            throw new RuntimeException("connect failed"+ future.cause().getMessage(), future.cause());
        } else{
            logger.error("connect failed");
            throw new RuntimeException("connect failed");
        }
        logger.info("connect success");*/
    }
    public Object doSend(Object o) throws Exception {
        return doSend(o, "0");
    }
    public Object doSend(Object o, String targetId) throws Exception {
        if (channel == null || !channel.isActive()) {
            synchronized(this) {
                wait(timeout);
            }
            if (channel == null || !channel.isActive()) {
                throw new RuntimeException("channel is not connect");
            }
        }
        String pid = UUID.randomUUID().toString().replaceAll("-", "");
        Message req = new Message();
        req.setBody(o);
        req.setId(pid);
        req.setTargetId(targetId);
        DefaultPackage pkg = new DefaultPackage(channel,req,timeout);
        channel.writeAndFlush(req);
        return pkg.get();
    }
    public void doDisConnect() throws Exception {
        if (channel != null) {
            channel.close();
            channel = null;
        }
    }
    public void doSync() throws Exception{
        try {
            //阻塞线程，直到连接关闭
            channel.closeFuture().sync();
        } finally {
            nioEventLoopGroup.shutdownGracefully();
        }
    }
    public void doClose() throws Exception {
        //优雅的退出，释放NIO线程组
        nioEventLoopGroup.shutdownGracefully();
    }

    public void setChannel(Channel channel) {
        this.channel = channel;
    }

    public Channel getChannel() {
        Channel c = channel;
        if (c == null || !c.isActive())
            return null;
        return c;
    }

    public int getTimeout() {
        return timeout;
    }
}